package com.fl.message.mq;

import com.alibaba.fastjson.JSON;
import com.fl.message.entity.UserMessage;
import com.fl.message.wb.OpinionSocketServer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.core.ExchangeTypes;
import org.springframework.amqp.rabbit.annotation.Exchange;
import org.springframework.amqp.rabbit.annotation.Queue;
import org.springframework.amqp.rabbit.annotation.QueueBinding;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Service;

import java.util.Collection;
import java.util.Collections;
import java.util.Objects;
import java.util.Random;
import java.util.concurrent.ConcurrentHashMap;
import java.util.stream.Collectors;
import java.util.stream.Stream;

import static com.fl.message.constant.MessageConstant.*;
import static com.fl.message.wb.OpinionSocketServer.*;

@Service
public class RabbitmqReceive {

    private static Logger logger = LoggerFactory.getLogger(RabbitmqReceive.class);

    @RabbitListener(bindings = @QueueBinding(value = @Queue(value = "public_queue_message", autoDelete = "false"),
            exchange = @Exchange(value = "public_exchange_message", type = ExchangeTypes.FANOUT)))
    public void consumer(String message) {
        UserMessage userMessage = JSON.parseObject(message, UserMessage.class);
        //未建立连线
        if (Objects.isNull(userMessage.getToId())) {
           //存在在线用户
            if (CUSTOMER.equals(userMessage.getType()) && !serverSet.isEmpty()){
                Random random = new Random();
                Object[] keys = serverSet.keySet().toArray();
                int index =  random.nextInt(serverSet.size());
                String key = String.valueOf(keys[index]);
                //绑定接收者
                userMessage.setToId(key);
                OpinionSocketServer opinionSocketServer = serverSet.get(key);
                //发送信息给接收者
//                opinionSocketServer.sendMessage2("11");
                opinionSocketServer.sendMessage(userMessage);
                //告知发送者
                OpinionSocketServer send = customerSet.get(userMessage.getFormId());
                userMessage.setMessage("正在查看您的消息，请稍等");
//                send.sendMessage2(userMessage.getMessage());
                send.sendMessage(userMessage);
                return;
            }
            //入库->待处理
            pendingMessage.add(userMessage);
            return;
        }
        if (CUSTOMER.equals(userMessage.getType())) {
            customer(userMessage);
        } else if (SERVER.equals(userMessage.getType())) {
            server(userMessage);
        }
    }

    public boolean consumerPendingMessage(UserMessage userMessage,String messageTips) {
        //判断当前接受者是否在线
        OpinionSocketServer sendMessage = customerSet.get(userMessage.getFormId());
        if (!Objects.isNull(sendMessage)) {
            userMessage.setMessage(messageTips);
            sendMessage.sendMessage(userMessage);
            return true;
        }
        return false;
    }

    public void server(UserMessage userMessage) {
        //判断当前接受者是否在线
        OpinionSocketServer receiveUser = customerSet.get(userMessage.getToId());
        if (!Objects.isNull(receiveUser)) {
            try {
                receiveUser.sendMessage(userMessage);
            } catch (Exception e) {
                OpinionSocketServer opinionSocketServer = serverSet.get(userMessage.getFormId());
                userMessage.setMessage("用户" + userMessage.getToId() + "已经断开链接");
                opinionSocketServer.sendMessage(userMessage);
                customerSet.remove(userMessage.getToId());
            }
        }
    }


    public void customer(UserMessage userMessage) {
        //判断当前接受者是否在线
        OpinionSocketServer receiveUser = serverSet.get(userMessage.getToId());
        if (!Objects.isNull(receiveUser)) {
            try {
                receiveUser.sendMessage(userMessage);
            } catch (Exception e) {
                reLink(userMessage);
            }
        }else {
            //消息入库
            pendingMessage.add(userMessage);
        }
    }

    //尝试再次发送
    public void reLink(UserMessage userMessage) {
        OpinionSocketServer receiveUser = serverSet.get(userMessage.getToId());
        if (!Objects.isNull(receiveUser)) {
            try {
                receiveUser.sendMessage(userMessage);
            } catch (Exception e) {
                dealError(userMessage);
            }
        }
        dealError(userMessage);
    }

    //执行错误处理
    public void dealError(UserMessage userMessage) {
        //移除用户
        serverSet.remove(userMessage.getToId());
        //消息入库
        pendingMessage.add(userMessage);
    }
}
